/*
 *   Copyright 2021 zzh
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */

package com.aduib.boot.common.disruptor;

import com.aduib.boot.common.concurrent.ThreadBuilder;
import com.aduib.boot.common.concurrent.ThreadFactoryBuilder;
import com.aduib.boot.common.exception.CommonException;
import com.aduib.boot.common.log.LOG;
import com.lmax.disruptor.FatalExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @description: Disruptors
 * @author: zzh
 * @date: 2021/9/16 11:21
 */
public final class Disruptors<E> {

  private static final int BUFFER_SIZE = 1024;
  RingBuffer<Elements<E>> ringBuffer;
  Disruptor<Elements<E>> disruptor;
  AtomicBoolean isStarted = new AtomicBoolean(false);
  private ThreadFactory threadFactory = ThreadFactoryBuilder.createJava("Disruptors", 5, true);
  private EventHandlerBuildStrategy strategy;
  private ElementEventTranslator<E> elementEventTranslator = new ElementEventTranslator<>();
  private ElementEventFactory<E> elementEventFactory = new ElementEventFactory<>();
  private ProducerType producerType = ProducerType.SINGLE;
  private WaitStrategy waitStrategy = new YieldingWaitStrategy();

  public static <T> Disruptors<T> builder() {
    return new Disruptors<>();
  }

  public Disruptors<E> setThreadFactory(ThreadFactory threadFactory) {
    this.threadFactory = threadFactory;
    return this;
  }

  public Disruptors<E> setElementEventTranslator(ElementEventTranslator<E> elementEventTranslator) {
    this.elementEventTranslator = elementEventTranslator;
    return this;
  }

  public Disruptors<E> setElementEventFactory(ElementEventFactory<E> elementEventFactory) {
    this.elementEventFactory = elementEventFactory;
    return this;
  }

  public Disruptors<E> setProducerType(ProducerType producerType) {
    this.producerType = producerType;
    return this;
  }

  public Disruptors<E> setWaitStrategy(WaitStrategy waitStrategy) {
    this.waitStrategy = waitStrategy;
    return this;
  }

  public Disruptors<E> setStrategy(EventHandlerBuildStrategy strategy) {
    this.strategy = strategy;
    return this;
  }

  /**
   * 添加事件处理器 默认使用并行处理策略
   *
   * @param eventHandler
   * @return
   */
  public Disruptors addDefaultEventHandler(ElementEventHandler<?> eventHandler) {
    if (Objects.isNull(strategy)) {
      strategy = new parallelHandlerBuildStrategy();
    }
    strategy.addWithOneHandler(eventHandler);
    return this;
  }

  public Disruptors addDefaultEventHandlers(List<ElementEventHandler<?>> eventHandler) {
    if (Objects.isNull(strategy)) {
      strategy = new parallelHandlerBuildStrategy();
    }
    strategy.addWithOneHandler(eventHandler);
    return this;
  }

  public Disruptors<E> start() {
    if (Objects.isNull(disruptor)) {
      throw new CommonException("disruptor is not created");
    }
    if (isStarted.get()) {
      throw new CommonException("disruptor already started");
    }
    disruptor.start();
    isStarted.getAndSet(true);
    getRingBuffer();
    LOG.d("Disruptor startUp");
    return this;
  }

  public Disruptors<E> stop() {
    if (Objects.isNull(disruptor)) {
      throw new CommonException("disruptor is not created");
    }
    if (!isStarted.get()) {
      throw new CommonException("disruptor is not started");
    }
    disruptor.shutdown();
    isStarted.getAndSet(false);
    return this;
  }

  public RingBuffer<Elements<E>> getRingBuffer() {
    if (Objects.isNull(disruptor)) {
      throw new CommonException("disruptor is not created");
    }
    if (!isStarted.get()) {
      throw new CommonException("disruptor is not started");
    }
    ringBuffer = disruptor.getRingBuffer();
    return ringBuffer;
  }

  public Disruptor<Elements<E>> getDisruptor() {
    return disruptor;
  }

  public Disruptors<E> create() {
    disruptor =
            new Disruptor<>(
                    elementEventFactory, BUFFER_SIZE, threadFactory, producerType, waitStrategy);
    disruptor.setDefaultExceptionHandler(new FatalExceptionHandler());
    //    disruptor.start();
    //    ringBuffer = disruptor.getRingBuffer();
    if (Objects.isNull(strategy)) {
      throw new IllegalArgumentException("EventHandlerBuildStrategy is not initialized");
    }
    strategy.buildHandlerList(this);
    return this;
  }

  public Disruptors<E> createAllOne() {
    create();
    start();
    Runtime.getRuntime()
        .addShutdownHook(
            ThreadBuilder.create(
                "disruptor-shutDownHook",
                () -> {
                  isStarted.compareAndSet(false, true);
                  this.stop();
                  LOG.d("Disruptor shutdown");
                }));
    return this;
  }

  public Disruptors<E> publishEvent(E e) {
    if (Objects.isNull(ringBuffer)) {
      throw new CommonException("disruptor is not created and started");
    }
    ringBuffer.publishEvent(elementEventTranslator, e);
    return this;
  }
}
